package kevin.hbase.monitor;

import kevin.hbase.util.EmailUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.ClusterStatus;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;

import java.io.IOException;
import java.util.Collection;

@Slf4j
public class HbaseMonitor extends Thread {
    private Connection connection;
    private HbaseStatus hbaseStatus = new HbaseStatus();
    private int consecutiveTime = 0;

    public HbaseMonitor(Connection connection) {
        this.connection = connection;
    }

    private void getRealHbaseStatus() throws IOException {
        hbaseStatus.reset();
        Admin admin = connection.getAdmin();
        ClusterStatus clusterStatus = admin.getClusterStatus();
        Collection<ServerName> aliveServerNames = clusterStatus.getServers();
        Collection<ServerName> deadServerNames = clusterStatus.getDeadServerNames();
        hbaseStatus.setAliveServers(aliveServerNames);
        hbaseStatus.setDeadServers(deadServerNames);
        int requestsCount = clusterStatus.getRequestsCount();
        hbaseStatus.setTotalRPS(requestsCount);
    }

    public boolean needReport() {
        return true;
    }

    public String createReport() {
        StringBuilder sb = new StringBuilder();
        sb.append("===========current total rps===========");
        sb.append("\n");
        sb.append(hbaseStatus.getTotalRPS());
        sb.append("\n");
        sb.append("===========dead regionserver info===========");
        sb.append("\n");
        for (ServerName deadServerName : hbaseStatus.getDeadServers()) {
            sb.append(deadServerName.toString());
            sb.append("\n");
        }
        sb.append("===========alive regionserver info===========");
        sb.append("\n");
        for (ServerName aliveServerName : hbaseStatus.getAliveServers()) {
            sb.append(aliveServerName.toString());
            sb.append("\n");
        }
        return sb.toString();
    }

    @Override
    public void run() {
        while (true) {
            try {
                getRealHbaseStatus();
                boolean need = needReport();
                String report = createReport();
                if (need) {
                    consecutiveTime++;
                    //
                    EmailUtil.sendMail("zhanggang@kauto.com", "Hbase Cluster Exception", report);
                    /*SmsDTO smsDTO = YunPianUtil.pushSms("13684969490",report);
                    System.out.println(smsDTO);*/
                } else {
                    consecutiveTime = 1;
                }
                log.info("Hbase Cluster Report>>\n" + report);
                int num = Math.min(30, consecutiveTime);
                sleep(120000 * num);
            } catch (Exception e) {
                e.printStackTrace();
                log.error(e.getMessage(), e);
            }
        }
    }

}
